package cc.fs.study.rxjava.util;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subjects.Subject;

/**
 * RxBus总事件线,所有的事件将会出现在这里
 * 根据订阅打了类型过滤消息通知
 * 暂时问题:容易导致内存泄露,即activity回收了订阅消息还能够接收,现在需要添加拦截
 */
public class RxBus {
    private static Subject<Object, Object> subject;

    //发射消息
    public static void send(Object object) {
        if (subject == null) {
            subject = PublishSubject.create();
        }
        subject.onNext(object);
    }

    //获取事件订阅
    public static Observable toObservable() {
        if (subject == null) {
            subject = PublishSubject.create();
        }
        return subject;
    }

    public static RxBus Builder() {
        return new RxBus();
    }

    @SuppressWarnings("unchecked")
    public <T> void subscribe(final Class<T> clas, Subscriber<T> subscriber) {
        toObservable()
//                .compose()
                .filter(new Func1<Object, Boolean>() {
            @Override
            public Boolean call(Object o) {
                if (clas.isInstance(o)) {
                    return true;
                }
                return false;
            }
        }).observeOn(Schedulers.io()).subscribe(subscriber);
    }
}
